package connector

import (
	"context"
	"fmt"
	"sync"
	"time"
)

// MemorySourceConnector represents an in-memory source connector
type MemorySourceConnector struct {
	// config is the connector configuration
	config map[string]interface{}
	// records are the records in memory
	records []*Record
	// position is the current position
	position int
	// metrics are the connector metrics
	metrics map[string]interface{}
	// mu is the mutex
	mu sync.RWMutex
	// started indicates if the connector is started
	started bool
}

// NewMemorySourceConnector creates a new in-memory source connector
func NewMemorySourceConnector() SourceConnector {
	return &MemorySourceConnector{
		records: make([]*Record, 0),
		metrics: make(map[string]interface{}),
	}
}

// Initialize initializes the connector
func (c *MemorySourceConnector) Initialize(config map[string]interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Store the configuration
	c.config = config

	// Check if data is provided
	if data, ok := config["data"]; ok {
		// Convert data to records
		if dataArray, ok := data.([]interface{}); ok {
			for _, item := range dataArray {
				if record, err := convertToRecord(item); err == nil {
					c.records = append(c.records, record)
				} else {
					return fmt.Errorf("failed to convert data item to record: %w", err)
				}
			}
		} else {
			return fmt.Errorf("data must be an array")
		}
	}

	// Initialize metrics
	c.metrics["records_total"] = len(c.records)
	c.metrics["records_read"] = 0
	c.metrics["start_time"] = time.Now().Unix()

	return nil
}

// Start starts the connector
func (c *MemorySourceConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already started
	if c.started {
		return fmt.Errorf("connector already started")
	}

	// Reset position
	c.position = 0
	c.started = true

	return nil
}

// Stop stops the connector
func (c *MemorySourceConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already stopped
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	c.started = false

	return nil
}

// Read reads records from the source
func (c *MemorySourceConnector) Read(ctx context.Context) (*RecordBatch, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return nil, fmt.Errorf("connector not started")
	}

	// Check if all records have been read
	if c.position >= len(c.records) {
		return nil, nil
	}

	// Get batch size
	batchSize := 10
	if size, ok := c.config["batch_size"].(int); ok {
		batchSize = size
	}

	// Read records
	end := c.position + batchSize
	if end > len(c.records) {
		end = len(c.records)
	}
	batch := c.records[c.position:end]
	c.position = end

	// Update metrics
	c.metrics["records_read"] = c.position
	c.metrics["last_read_time"] = time.Now().Unix()

	return NewRecordBatch(batch), nil
}

// Commit commits the offset
func (c *MemorySourceConnector) Commit(ctx context.Context, offset interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Check if offset is valid
	if offset == nil {
		return nil
	}

	// Convert offset to int
	pos, ok := offset.(int)
	if !ok {
		return fmt.Errorf("invalid offset type")
	}

	// Check if offset is valid
	if pos < 0 || pos > len(c.records) {
		return fmt.Errorf("invalid offset value")
	}

	// Update position
	c.position = pos

	return nil
}

// GetMetrics gets connector metrics
func (c *MemorySourceConnector) GetMetrics() map[string]interface{} {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy metrics
	metrics := make(map[string]interface{})
	for k, v := range c.metrics {
		metrics[k] = v
	}

	return metrics
}

// MemorySinkConnector represents an in-memory sink connector
type MemorySinkConnector struct {
	// config is the connector configuration
	config map[string]interface{}
	// records are the records in memory
	records []*Record
	// metrics are the connector metrics
	metrics map[string]interface{}
	// mu is the mutex
	mu sync.RWMutex
	// started indicates if the connector is started
	started bool
}

// NewMemorySinkConnector creates a new in-memory sink connector
func NewMemorySinkConnector() SinkConnector {
	return &MemorySinkConnector{
		records: make([]*Record, 0),
		metrics: make(map[string]interface{}),
	}
}

// Initialize initializes the connector
func (c *MemorySinkConnector) Initialize(config map[string]interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Store the configuration
	c.config = config

	// Initialize metrics
	c.metrics["records_total"] = 0
	c.metrics["start_time"] = time.Now().Unix()

	return nil
}

// Start starts the connector
func (c *MemorySinkConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already started
	if c.started {
		return fmt.Errorf("connector already started")
	}

	c.started = true

	return nil
}

// Stop stops the connector
func (c *MemorySinkConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already stopped
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	c.started = false

	return nil
}

// Write writes records to the sink
func (c *MemorySinkConnector) Write(ctx context.Context, batch *RecordBatch) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Check if batch is valid
	if batch == nil || len(batch.Records) == 0 {
		return nil
	}

	// Write records
	c.records = append(c.records, batch.Records...)

	// Update metrics
	c.metrics["records_total"] = len(c.records)
	c.metrics["last_write_time"] = time.Now().Unix()

	return nil
}

// Flush flushes any buffered records
func (c *MemorySinkConnector) Flush(ctx context.Context) error {
	// No buffering in memory sink
	return nil
}

// GetMetrics gets connector metrics
func (c *MemorySinkConnector) GetMetrics() map[string]interface{} {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy metrics
	metrics := make(map[string]interface{})
	for k, v := range c.metrics {
		metrics[k] = v
	}

	return metrics
}

// GetRecords gets the records
func (c *MemorySinkConnector) GetRecords() []*Record {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy records
	records := make([]*Record, len(c.records))
	copy(records, c.records)

	return records
}

// convertToRecord converts an interface to a record
func convertToRecord(data interface{}) (*Record, error) {
	// Check if data is a map
	dataMap, ok := data.(map[string]interface{})
	if !ok {
		return nil, fmt.Errorf("data must be a map")
	}

	// Create a new record
	record := NewRecord(nil, nil)

	// Set key if provided
	if key, ok := dataMap["id"]; ok {
		if keyStr, ok := key.(string); ok {
			record.Key = []byte(keyStr)
		}
	}

	// Set value
	record.Value = []byte(fmt.Sprintf("%v", data))

	return record, nil
}
